package com.bw.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.app.functions.DimAsyncFunction;
import com.bw.bean.OrderDetail;
import com.bw.bean.OrderInfo;
import com.bw.bean.OrderWide;
import com.bw.common.GmallConfig;
import com.bw.utils.DateTimeUtil;
import com.bw.utils.DimUtil;
import com.bw.utils.JdbcUtil;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.concurrent.TimeUnit;

public class OrderWideApp {
    // 业务数据源 app -->nginx-->springboot---->mysql(业务库)--->FlinkCDC--->Kafka(ods)--->Flink(base_db_app)----Kafka(dwd)/Hbase(dim)---Flink
    //  程序员             jar ---->FlinkCDC---base_db_app---OrderWideApp
    public static void main(String[] args) throws Exception {
        // 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //        开启CK，创建文件夹
//        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:9820/gmall-flink/ck"));
//         每隔5秒保存一次
//        env.enableCheckpointing(5000L);
//        精准消费一次
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        配置超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(100000L);
//        同时最大可以运行几个ck
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
//        上一个ck的尾部到下一个ck的头之间间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
        // env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 2.读取Kafka dwd_order_info数据，并转成Java实体对象1, 生成水印

        String orderInfoSourceTopic = "dwd_order_info";
        String orderDetailSourceTopic = "dwd_order_detail";
        String orderWideSinkTopic = "dwm_order_wide";
        String groupId = "order_wide_group_2105b_03";
        SingleOutputStreamOperator<OrderInfo> orderInfoDS = env.addSource(MyKafkaUtil.getKafkaConsumer(orderInfoSourceTopic, groupId))
                .map(new MapFunction<String, OrderInfo>() {
                    @Override
                    public OrderInfo map(String value) throws Exception {
                        // 转成Java Bean
                        OrderInfo orderInfo = JSON.parseObject(value, OrderInfo.class);
                        String create_time = orderInfo.getCreate_time();
                        // ["2020-12-04" "23:28:50"]
                        String[] dateArr = create_time.split(" ");
                        orderInfo.setCreate_date(dateArr[0]);
                        orderInfo.setCreate_hour(dateArr[1].split(":")[0]);
                        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        orderInfo.setCreate_ts(simpleDateFormat.parse(create_time).getTime());
                        return orderInfo;
                    }
                    // forMonotonousTimestamps 升序
                }).assignTimestampsAndWatermarks(WatermarkStrategy.<OrderInfo>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<OrderInfo>() {
                    @Override
                    public long extractTimestamp(OrderInfo orderInfo, long l) {
                        return orderInfo.getCreate_ts();
                    }
                }));

        // 3.读取Kafka dwd_order_detail数据，并转成Java实体对象2
        SingleOutputStreamOperator<OrderDetail> orderDetailDS = env.addSource(MyKafkaUtil.getKafkaConsumer(orderDetailSourceTopic, groupId)).map(new MapFunction<String, OrderDetail>() {
            @Override
            public OrderDetail map(String value) throws Exception {
                OrderDetail orderDetail = JSON.parseObject(value, OrderDetail.class);
                String create_time = orderDetail.getCreate_time();
                SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                orderDetail.setCreate_ts(simpleDateFormat.parse(create_time).getTime());
                return orderDetail;
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<OrderDetail>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<OrderDetail>() {
            @Override
            public long extractTimestamp(OrderDetail orderDetail, long l) {
                return orderDetail.getCreate_ts();
            }
        }));
        // 4.双流Join Java实体对象1和Java实体对象2变成一个Java实体对象3

        SingleOutputStreamOperator<OrderWide> orderWideNoDimDS = orderInfoDS.keyBy(new KeySelector<OrderInfo, Long>() {
            @Override
            public Long getKey(OrderInfo orderInfo) throws Exception {
                return orderInfo.getId();

            }
        }).intervalJoin(orderDetailDS.keyBy(new KeySelector<OrderDetail, Long>() {
            @Override
            public Long getKey(OrderDetail orderDetail) throws Exception {
                return orderDetail.getOrder_id();
            }
        })).between(Time.seconds(-5), Time.seconds(5)).process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
            @Override
            public void processElement(OrderInfo orderInfo, OrderDetail orderDetail, Context context, Collector<OrderWide> collector) throws Exception {
                collector.collect(new OrderWide(orderInfo, orderDetail));
            }
        });
        // 5.关联Hbase里面的维度表.比如用户、地区，商品 品牌、分类、squ
        // 5.1 先关联用户

//        SingleOutputStreamOperator<OrderWide> userDS = orderWideNoDimDS.map(new MapFunction<OrderWide, OrderWide>() {
//            @Override
//            public OrderWide map(OrderWide orderWide) throws Exception {
//                String user_id = orderWide.getUser_id() + "";
//                Class.forName(GmallConfig.PHOENIX_DRIVER);
//                Connection connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
//                JSONObject dim_user_info = DimUtil.getDimInfo(connection, "DIM_USER_INFO", user_id);
//                return orderWide;
//            }
//        });
        SingleOutputStreamOperator<OrderWide> orderWideWithUserDS = AsyncDataStream.
                unorderedWait(orderWideNoDimDS, new DimAsyncFunction<OrderWide>("DIM_USER_INFO") {
                    @Override
                    public String getKey(OrderWide input) {
                        return input.getUser_id().toString();
                    }

                    @Override
                    public void join(OrderWide orderWide, JSONObject dimInfo) throws ParseException {
                        orderWide.setUser_gender(dimInfo.getString("GENDER"));
                        String birthday = dimInfo.getString("BIRTHDAY");
                        int age = DateTimeUtil.getAge(birthday);
//                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
//                        long currentTs = System.currentTimeMillis();
//                        long ts = sdf.parse(birthday).getTime();
//                        long age = (currentTs - ts) / (1000 * 60 * 60 * 24 * 365L);
                        orderWide.setUser_age(age);

                    }
                }, 60, TimeUnit.SECONDS);
        // 5.2  关联地区
        SingleOutputStreamOperator<OrderWide> orderWideWithProvinceDS = AsyncDataStream.unorderedWait(orderWideWithUserDS,
                new DimAsyncFunction<OrderWide>("DIM_BASE_PROVINCE") {
                    @Override
                    public String getKey(OrderWide orderWide) {
                        return orderWide.getProvince_id().toString();
                    }

                    @Override
                    public void join(OrderWide orderWide, JSONObject dimInfo) throws ParseException {
                        orderWide.setProvince_name(dimInfo.getString("NAME"));
                        orderWide.setProvince_area_code(dimInfo.getString("AREA_CODE"));
                        orderWide.setProvince_iso_code(dimInfo.getString("ISO_CODE"));
                        orderWide.setProvince_3166_2_code(dimInfo.getString("ISO_3166_2"));
                    }
                }, 120, TimeUnit.SECONDS);

        //5.3 关联SKU维度
        SingleOutputStreamOperator<OrderWide> orderWideWithSkuDS = AsyncDataStream.unorderedWait(
                orderWideWithProvinceDS, new DimAsyncFunction<OrderWide>("DIM_SKU_INFO") {
                    @Override
                    public void join(OrderWide orderWide, JSONObject jsonObject) throws ParseException {
                        orderWide.setSku_name(jsonObject.getString("SKU_NAME"));
                        orderWide.setCategory3_id(jsonObject.getLong("CATEGORY3_ID"));
                        orderWide.setSpu_id(jsonObject.getLong("SPU_ID"));
                        orderWide.setTm_id(jsonObject.getLong("TM_ID"));
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return String.valueOf(orderWide.getSku_id());
                    }
                }, 120, TimeUnit.SECONDS);

//        //5.4 关联SPU维度
        SingleOutputStreamOperator<OrderWide> orderWideWithSpuDS = AsyncDataStream.unorderedWait(
                orderWideWithSkuDS, new DimAsyncFunction<OrderWide>("DIM_SPU_INFO") {
                    @Override
                    public void join(OrderWide orderWide, JSONObject jsonObject) throws ParseException {
                        orderWide.setSpu_name(jsonObject.getString("SPU_NAME"));
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return String.valueOf(orderWide.getSpu_id());
                    }
                }, 120, TimeUnit.SECONDS);
//
//        //5.5 关联TM维度
        SingleOutputStreamOperator<OrderWide> orderWideWithTmDS = AsyncDataStream.unorderedWait(
                orderWideWithSpuDS, new DimAsyncFunction<OrderWide>("DIM_BASE_TRADEMARK") {
                    @Override
                    public void join(OrderWide orderWide, JSONObject jsonObject) throws ParseException {
                        orderWide.setTm_name(jsonObject.getString("TM_NAME"));
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return String.valueOf(orderWide.getTm_id());
                    }
                }, 60, TimeUnit.SECONDS);
//
//        //5.6 关联Category维度
        SingleOutputStreamOperator<OrderWide> orderWideWithCategory3DS = AsyncDataStream.unorderedWait(
                orderWideWithTmDS, new DimAsyncFunction<OrderWide>("DIM_BASE_CATEGORY3") {
                    @Override
                    public void join(OrderWide orderWide, JSONObject jsonObject) throws ParseException {
                        orderWide.setCategory3_name(jsonObject.getString("NAME"));
                    }

                    @Override
                    public String getKey(OrderWide orderWide) {
                        return String.valueOf(orderWide.getCategory3_id());
                    }
                }, 120, TimeUnit.SECONDS);


        // 6.写入Kafka对应的主题当中
        orderWideWithCategory3DS.print(">>>>>");
        orderWideWithCategory3DS.map(JSONObject::toJSONString).addSink(MyKafkaUtil.getKafkaProducer(orderWideSinkTopic));
        // 7.执行任务
        env.execute("OrderWideApp");

    }
}
